package com.example.mq.server;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.server.core.*;
import com.example.mq.server.datacenter.DiskDataCenter;
import com.example.mq.server.datacenter.MemoryDataCenter;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/*
* 虚拟主机
* 每个虚拟主机下面都管理着自己的 交换机 队列 消息 绑定 数据
* 同时提供api供上层调用
* */
@Slf4j
@Data
public class VirtualHost {
    private final String virtualHostName;
    private final MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private final DiskDataCenter diskDataCenter = new DiskDataCenter();
    private final Router router = new Router();
    // 针对交换机操作进行加锁
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();
    private ConsumerManager consumerManager = new ConsumerManager(this);


    public VirtualHost(String name) throws MqException {
        virtualHostName = name;

        // 对于memoryDataCenter来说 不需要进行初始化 只需要new出来一个对象即可
        // 但是对于diskDataCenter 需要进行初始化操作,针对数据库进行建表和初始化数据
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MqException | IOException | ClassNotFoundException e) {
            e.printStackTrace();
            log.info("[VirtualHost] 恢复内存数据失败!");
        }
    }

    /**
     * 创建交换机,不存在就创建,如果存在直接返回
     * @param exchangeName 交换机名字 约定 : 虚拟主机名+交换机名
     * @param type 交换机类型
     * @param durable 是否进行持久化
     * @param autoDelete 是否自动删除
     * @param arguments 额外参数
     * @return
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable, boolean autoDelete,
                                   Map<String,Object> arguments){
        // 把交换机的名字加上虚拟主机前缀
        exchangeName = virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 判断虚拟主机是否存在,通过内存直接进行查询
                Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
                if(existExchange != null){
                    log.info("[VirtualHost] 交换机已经存在! exchangeName = "+ exchangeName);
                    return true;
                }

                // 交换机不存在,真正创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setType(type);
                exchange.setArguments(arguments);

                // 写入硬盘
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }

                // 写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("[VirtualHost] 交换机创建完成! exchangeName = "+exchangeName);
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();log.info("[VirtualHost] 交换机创建失败! exchangeName = "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName
     * @return
     */
    public boolean deleteExchange(String exchangeName){
        exchangeName = virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1.先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null){
                    throw new MqException("[VirtualHost] 交换机不存在,无法进行删除!");
                }
                // 2.删除交换机
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(toDelete);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("[VirtualHost] 交换机删除成功! exchangeName = "+exchangeName);
                return true;
            }
        }catch(Exception e){
            e.printStackTrace();
            log.info("[VirtualHost] 交换机删除失败! exchangeName = "+exchangeName);
            return false;
        }
    }

    /**
     * 创建队列
     * @param queueName 用户指定的队列名,非实际队列名
     * @param durable 持久化
     * @param exclusive 是否独占
     * @param autoDelete 是否在不使用时自动删除
     * @param arguments 额外参数
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String,Object> arguments){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker) {
                // 1.查询是否存在
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                if(existQueue != null){
                    log.info("[VirtualHost] 队列已经存在! queueName = "+queueName);
                    return true;

                }
                // 2.队列不存在正式创建
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3.插入硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                // 4.插入内存
                memoryDataCenter.insertQueue(queue);
                log.info("[VirtualHost] 队列创建成功! queueName = "+queueName);
            }
            return true;
        }catch(Exception e){
            e.printStackTrace();
            log.info("[VirtualHost] 队列创建失败! queueName = "+queueName);
            return false;
        }
    }

    /**
     * 删除队列
     * @param queueName
     * @return
     */
    public boolean queueDelete(String queueName){
        queueName = virtualHostName+queueName;
        try{
            synchronized (queueLocker) {
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                if(existQueue == null){
                    throw new MqException("[VirtualHost] 队列不存在,无法删除!");
                }
                // 删除硬盘数据
                if(existQueue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                // 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                log.info("[VirtualHost] 队列删除成功! queueName = "+queueName);
            }
            return true;
        }catch(Exception e){
            log.info("[VirtualHost] 队列删除失败! queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 队列绑定
     * @param exchangeName
     * @param queueName
     * @param bindingKey
     * @return
     */
    public boolean queueBind(String exchangeName, String queueName, String bindingKey){
        exchangeName = virtualHostName+exchangeName;
        queueName = virtualHostName+queueName;
        try{
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existBinding != null){
                        throw new MqException("[VirtualHost] queueBind失败! exchangeName = "+exchangeName+
                                " queueName = "+queueName);
                    }

                    // 1.验证bindingKey是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey不合法! bindingKey = "+bindingKey);
                    }

                    // 2.创建binding对象
                    Binding binding = new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);

                    // 3.获取一下交换机和队列,如果其中有一个不存在,这样的绑定也是无法创建的
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在,无法创建绑定! exchangeName = "+exchangeName);
                    }

                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[VirtualHost] 队列不存在,无法创建绑定! queueName = "+queueName);
                    }
                    // 4.写入
                    if(queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    log.info("[VirtualHost] queueBind成功!");
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 解除绑定
     * @param exchangeName
     * @param queueName
     * @return
     */
    public boolean queueUnBind(String exchangeName, String queueName){
        exchangeName = virtualHostName+exchangeName;
        queueName = virtualHostName+queueName;
        try{
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existBinding == null){
                        throw new MqException("[VirtualHost] 绑定不存在,无法解除绑定! exchangeName = "+exchangeName+" queueName = "+queueName);
                    }
                    diskDataCenter.deleteBinding(existBinding);
                    memoryDataCenter.deleteBinding(existBinding);
                    log.info("[VirtualHost] queueUnBind成功!");
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发送消息到指定队列
     * @param exchangeName 交换机
     * @param routingKey 和bindingKey匹配
     * @param basicProperties 基本属性 messageId,routingKey,deliverMode
     * @param body 消息体
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try{
            // 1.拼接虚拟主机和交换机的名字
            exchangeName = virtualHostName+exchangeName;
            // 2.判断RoutingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] RoutingKey不合法! routingKey = "+routingKey);
            }
            // 3.查看交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[virtualHost] 交换机不存在! exchangeName = "+exchangeName);
            }
            // 4.判定交换机的类型
            if(exchange.getType() == ExchangeType.DIRECT){
                // 以routingKey作为队列的名字,写入指定的队列中
                String queueName = virtualHostName+routingKey;
                // 5.创建消息
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                // 6.查找队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在! queueName = "+queueName);
                }
                sendMessage(queue,message);
            }else{
                // 找到该交换机关联的所有绑定,并遍历这些绑定
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBinding(exchangeName);
                for(Map.Entry<String,Binding> entry :bindingsMap.entrySet()){
                    Binding bind = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(bind.getQueueName());
                    if(queue == null){
                        continue;
                    }
                    // 构造消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    // 判断转发规则
                    if(!router.route(exchange.getType(),bind,message)){
                        continue;
                    }
                    // 发送消息
                    sendMessage(queue,message);
                }
            }

            return true;
        }catch(Exception e){
            e.printStackTrace();
            return false;
        }

    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // deliverMode 为1表示不持久化 为2表示持久化
        if(message.getDeliverMode() == 2){
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);

        // 通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者,当一个队列收到消息之后,就把这个消息发送给对应的订阅者
     * @param consumerTag 消费者的身份标识
     * @param queueName 队列名
     * @param autoAck 消息被消费完成之后的应答方式 true表示自动应答 false表示手动应答
     * @param consumer 回调函数 类型是函数式接口 这样后续调用basicConsume传实参的时候,就可以使用Lambda的方式了
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        // 构造一个consumerEnv对象 找到其对应的队列 将消息放入队列
        queueName = virtualHostName+queueName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            log.info("[virtualHost] basicConsume成功! queueName = "+queueName);
            return true;
        }catch (Exception e){
            log.info("[virtualHost] basicConsume失败! queueName = "+queueName);
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName = virtualHostName+queueName;
        try{
            // 1.删除硬盘上的数据
            Message message = memoryDataCenter.getMessageWaitAck(queueName,messageId);
            if(message == null){
                throw new MqException("[virtualHost] 要确认的消息不存在! messageId = "+messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[virtualHost] 对列不存在! queueName = "+queueName);
            }
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            // 2.删除待确认队列的消息
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            // 3.删除内存中的消息
            memoryDataCenter.removeMessage(messageId);
            log.info("[virtualHost] 消息确认成功! queueName = "+queueName +" messageId = "+messageId);
            return true;
        } catch (Exception e) {
            log.info("[virtualHost] 消息确认失败! queueName = "+queueName +" messageId = "+messageId);
            e.printStackTrace();
            return false;
        }
    }
}
